﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Linq;
using System.Text;
using System.Threading.Actors.Languages;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors
{
    // For testing.
    public class StandaloneActorState : IActorState, IPersistenceHandler
    {

        private static readonly Task s_preCompletedTask = Task.FromResult(true);

        // Here's the mental model for partitions.  A service is created with 1 or more partitions.  
        // If it is multiple partitions, we create a unique IActorState for the number of partitions.
        // Actors then specify a partition name, and we will map that onto a particular partition 
        // number.  All actors that get mapped to the same partition will share the same IActorState.
        // This lets us co-locate multiple actors on one WF service instance.
        //
        // _HOWEVER_, THIS IS NOT WHAT WE DO IN TESTING.  One challenge with actor development is to 
        // ensure that your actor scopes all of its field names with an appropriate actor instance name.
        // Failing to do so means that other actors in the same IActorState will collide, and pollute
        // each other's state.  This can happen in the real world, so let's make that happen in our
        // StandaloneActorState all the time to increase the chance of seeing failures early in development.
        //
        // For our partitioned actor state lookup, a more realistic object model would use the actor 
        // name and a partition number as the key in a dictionary.  A theoretically idealized model
        // would create a unique StandaloneActorState for every unique partition name, but
        // we're not building that system in the real world.  To help you find bugs in your actor methods,
        // we'll use only the actor name.  Use helpers to do the lookup, in case we change the 
        // implementation in the future to the more realistic model above.  But in the mean time,
        // we'll store the 1 aggregate StandaloneActorState for all partitioned actors in the actor directory,
        // just like everything else.
        private static Dictionary<string, StandaloneActorState> s_actorStateDirectory = new Dictionary<string, StandaloneActorState>();

        //private static Dictionary<string, StandaloneActorState> s_partitionedActorStateDirectory = new Dictionary<string, StandaloneActorState>();

        private static Dictionary<string, StandaloneActor> s_actorDirectory = new Dictionary<string, StandaloneActor>(); 

        private readonly PubSubHelper m_pubSubHelper;
        private readonly MethodCallHelper m_methodCallHelper;
        private AssemblyHelper m_assemblyHelper;

        // Realistically, this is unneeded logic for the unit tests.
        // However, it's nice to test StateBufferingHelper as part of the unit tests.
        private readonly StateBufferingHelper m_stateHelper;
        private readonly ConcurrentDictionary<string, object> m_permaStore;

        // For a particular method call, did the actor method write anything?
        private bool m_didWritesOccur = false;

        private const string m_myPartitionName = "";

        private readonly SemaphoreSlim m_asyncStateLock = new SemaphoreSlim(1, 1);

        public StandaloneActorState(string actorName = null, int numPartitions = 1)
        {
            if (actorName != null)
            {
                // The numPartitions parameter lets us do something different with partitioned actors 
                // in the future, if we wish.
                lock (s_actorStateDirectory) { s_actorStateDirectory[actorName] = this; }
            }

            m_pubSubHelper = new PubSubHelper(this, m_myPartitionName);
            m_methodCallHelper = new MethodCallHelper(this);
            m_stateHelper = new StateBufferingHelper(this);
            m_permaStore = m_stateHelper.GetPermaStore(m_myPartitionName);
            m_assemblyHelper = new AssemblyHelper(this, m_permaStore, m_methodCallHelper);
            m_assemblyHelper.AddAssemblyResolver();
        }

        public SemaphoreSlim AsyncStateLock { get { return m_asyncStateLock; } }

        public void Dispose()
        {
            if (m_assemblyHelper != null) m_assemblyHelper.RemoveAssemblyResolver();
            if(m_stateHelper != null) m_stateHelper.Dispose();
            if(m_pubSubHelper != null) m_pubSubHelper.Close();
        }

        public string PartitionName {
            get { return m_myPartitionName; }
        }

        // For debugging
        public int NumItemsInState {
            get { return m_permaStore.Count; }
        }

        public static bool TryLookupActorState(string actorName, string partitionName, out StandaloneActorState retrievedState)
        {
            // If we change how partitioned actors are stored, the partition name may become relevant.
            lock (s_actorStateDirectory)
                return s_actorStateDirectory.TryGetValue(actorName, out retrievedState);
        }

        internal static bool TryLookupActor(string actorName, out StandaloneActor retrievedActor)
        {
            lock (s_actorDirectory)
                return s_actorDirectory.TryGetValue(actorName, out retrievedActor);
        }

        public void Set(String key, Object value)
        {
            m_didWritesOccur = true;
            m_stateHelper.Set(m_myPartitionName, key, value);
        }

        public Object Get(String key)
        {
            return m_stateHelper.Get(m_myPartitionName, key);
        }

        public bool TryGet(String key, out object value)
        {
            return m_stateHelper.TryGet(m_myPartitionName, key, out value);
        }

        public void Remove(String key)
        {
            m_didWritesOccur = true;
            m_stateHelper.Remove(m_myPartitionName, key);
        }

        public Task Flush()
        {
            return m_stateHelper.Flush(m_myPartitionName);
        }

        // This method removes the prefix from those strings.
        public IEnumerable<String> GetAllKeysWithPrefix(String keyPrefix)
        {
            var filteredKeys = m_stateHelper.GetAllKeysWithPrefix(m_myPartitionName, keyPrefix);

            // For testing, let's reverse the keys.  We provide no ordering guarantee here.
            filteredKeys = filteredKeys.Reverse();
            return filteredKeys;
        }

        public Object CallMethod(string methodName, Object[] parameters)
        {
            m_didWritesOccur = false;
            return m_methodCallHelper.CallMethod(methodName, parameters);
        }

        public void Subscribe(IPublicationCallbackContract callerProxy, String eventType)
        {
            m_pubSubHelper.Subscribe(callerProxy, eventType);
        }

        public void Unsubscribe(IPublicationCallbackContract callerProxy, String eventType)
        {
            m_pubSubHelper.Unsubscribe(callerProxy, eventType);
        }

        public void UnsubscribeAll(IPublicationCallbackContract callerProxy)
        {
            if (callerProxy == null)
                m_pubSubHelper.UnsubscribeEverything();
            else
                m_pubSubHelper.UnsubscribeAll(callerProxy);
        }

        public void Publish(String eventType, object eventPayload)
        {
            m_pubSubHelper.Publish(eventType, eventPayload);
        }

        public void Publish(string eventType, object eventPayload, params object[] eventPayloadQualifiers)
        {
            m_pubSubHelper.Publish(eventType, eventPayload, eventPayloadQualifiers);
        }


        public bool ChangesPendingForCurrentOperation()
        {
            return m_didWritesOccur;
        }


        public void ClearChangesForCurrentOperation()
        {
            // do nothing
        }


        public IActorProxy GetActorProxy(string externalActorName, string partitionName = null)
        {
            return new FakeActorProxy(externalActorName, this, partitionName);
        }


        public string GetId()
        {
            return "fabric:/actor/someActor";
        }


        public void AddLanguagePack(SupportedLanguages language, Func<ILanguagePackage> pack)
        {
            m_methodCallHelper.AddLanguagePack(language, pack);
        }

        public void AddMethod(SupportedLanguages language, string methodName, string methodImpl)
        {
            m_methodCallHelper.AddMethod(language, methodName, methodImpl);
        }


        public bool CreateActor(string actorName, string creationString, int numPartitions = 1)
        {
            if (s_actorStateDirectory.ContainsKey(actorName)) return true;

            StandaloneActorState state = new StandaloneActorState(actorName, numPartitions); // will get registered in s_actorDirectory
            new StandaloneActor(state, actorName, null, null);
            return false;
        }

        // For supporting existing actors created on explicitly created state objects (unit testing)
        public bool AddExistingActor(string actorName, StandaloneActor actor, StandaloneActorState state)
        {
            lock (s_actorStateDirectory) { s_actorStateDirectory[actorName] = state; }
            lock (s_actorDirectory)
                s_actorDirectory[actorName] = actor;

            return false;
        }

        public void DeleteActor(string actorName)
        {
            lock (s_actorStateDirectory)
            {
                if (s_actorStateDirectory.ContainsKey(actorName))
                    s_actorStateDirectory.Remove(actorName);
            }
			
			lock (s_actorDirectory)
			{
			    if (s_actorDirectory.ContainsKey(actorName))
				    s_actorDirectory.Remove(actorName);
			}
        }

        // Simply copy all cached operations to the permanent store.
        Task IPersistenceHandler.OnPersistAsync(IList<ChangeOperation> toPersist)
        {
            foreach (var operation in toPersist)
            {
                var partitionKey = operation.PartitionName;
                if (operation.OperationType == ChangeOperationType.Set)
                {
                    // Look for assemblies - this is where we will scour for methods, handle assembly references, etc.
                    // CheckForAssemblyOrMethod will add anything necessary to our state - no need to add it to our 
                    // permastore here, if it's an assembly.
                    if (!m_assemblyHelper.CheckForAssemblyOrMethod(operation.OperationKey, operation.OperationValue, scour: true))
                        m_stateHelper.GetPermaStore(partitionKey)[operation.OperationKey] = operation.OperationValue;
                }
                else if (operation.OperationType == ChangeOperationType.Remove)
                {
                    object temp;
                    m_permaStore.TryRemove(operation.OperationKey, out temp);
                }
                else throw new InvalidOperationException("Unknown CachedOperation type");
            }

            return s_preCompletedTask;
        }

        public void Configure(string configurationString)
        {
            // We don't have ETW support for StandaloneActorState.  So no need to configure that part.
            throw new NotImplementedException("Setting configuration NYI in StandaloneActorState");
        }

        public IActorPromise GetActorProxyAsync(string externalActorName, string partitionName = null)
        {
            throw new NotImplementedException();
        }
    } // StandaloneActorState

    class FakeActorProxy : IActorProxy
    {
        private string m_otherActorName;
        private string m_otherActorPartitionName;
        private IActorState m_myState;
        private const string m_myPartitionKey = "";

        public FakeActorProxy(string otherActorName, IActorState myState, string partitionName = null)
        {
            Contract.Requires(otherActorName != null);
            Contract.Requires(myState != null);

            m_otherActorName = otherActorName;
            m_otherActorPartitionName = partitionName;
            m_myState = myState;
        }

        public IActorStateObservable this[string eventType]
        {
            get { return new FakeObservable(); }
        }

        public void Command(string methodName, object[] parameters)
        {
            StandaloneActorState otherActorState = null;

            if (StandaloneActorState.TryLookupActorState(m_otherActorName, m_otherActorPartitionName, out otherActorState))
            {
                otherActorState.CallMethod(methodName, parameters);
            }
            else throw new InvalidOperationException("Could not find actor named " + m_otherActorName);
        }

        public IActorPromise Request(string methodName, object[] parameters)
        {
            StandaloneActorState otherActorState = null;
            StandaloneActor otherActor = null;

            StandaloneActorState.TryLookupActorState(m_otherActorName, m_otherActorPartitionName, out otherActorState);

            // Try calling methods on a StandaloneActorState implementation to exercise the full code path.
            StandaloneActorState.TryLookupActor(m_otherActorName, out otherActor);

            if (otherActorState != null)
            {
                var task = Task.Factory.StartNew(() =>
                    {
                        if (otherActor != null)
                        {
                            String[] jsonParameters = new String[parameters.Length];
                            for (int i = 0; i < jsonParameters.Length; i++)
                                jsonParameters[i] = SerializationHelper.Serialize(parameters[i]);
                            // IActor's CallMethod returns a String containing an ActorServiceResult in JSON.
                            var jsonResult = otherActor.CallMethod("FakeActorProxy", -1, m_otherActorPartitionName, methodName, jsonParameters).Result;
                            var result = ActorServiceResult.FromJson(jsonResult);
                            if (result.Status == ActorServiceResultStatus.Succeeded)
                                return result.Result;
                            else
                            {
                                Contract.Assert(result.Status == ActorServiceResultStatus.Failed);
                                throw result.ConstructException();
                            }
                        }

                        // IActorState's CallMethod returns an object directly - just serialize that & return it.
                        return SerializationHelper.Serialize(otherActorState.CallMethod(methodName, parameters));
                    });
                return new ActorPromise(m_myState, task);
            }

            else throw new InvalidOperationException("Could not find actor named " + m_otherActorName);
        }

        public bool AddAssembly(string assemblyName, byte[] assemblyBytes)
        {
            StandaloneActor otherActor = null;
            StandaloneActorState.TryLookupActor(m_otherActorName, out otherActor);

            if (otherActor != null)
            {
                String response = otherActor.AddAssembly(-1, m_otherActorPartitionName, assemblyName, assemblyBytes).Result;
                ActorServiceResult asr = ActorServiceResult.FromJson(response);
                if (asr.Status != ActorServiceResultStatus.Succeeded)
                {
                    throw asr.ConstructException();
                }
                return true;
            }
            else throw new InvalidOperationException("Could not find actor named " + m_otherActorName);
        }
    } // FakeActorProxy


    class FakeObservable : IActorStateObservable
    {

        public IDisposable Subscribe(string eventHandlerName)
        {
            return new FakeDisposable();
        }

        public void Unsubscribe(string eventHandlerName)
        {
        }
    }

    class FakeDisposable : IDisposable
    {

        public void Dispose()
        {
        }
    }
}
